package com.ylkj.bus.onoffline.report;

import cn.hutool.core.date.DateTime;
import cn.hutool.core.date.DateUtil;
import cn.hutool.core.util.StrUtil;
import com.alibaba.fastjson2.JSON;
import com.alibaba.fastjson2.JSONObject;
import com.hwl.redis.RedisConfig;
import com.ylkj.bus.MessageContent;
import com.ylkj.bus.device.report.mode.DeviceMoDO;
import com.ylkj.bus.device.report.mode.ProductDO;
import com.ylkj.bus.monitor.equipment.model.EquipmentAlarmMonitorLogEnty;
import com.ylkj.bus.onoffline.monitor.rule.code.RuleReportProxyEnum;
import com.ylkj.bus.onoffline.report.constant.DynamicsPropertyConstant;
import com.ylkj.bus.onoffline.report.constant.EquipmentAlarmConstant;
import com.ylkj.bus.onoffline.report.constant.EquipmentProxyEnum;
import com.ylkj.bus.onoffline.report.constant.MsgKeyConstant;
import com.ylkj.dao.RdsConfig;
import com.ylkj.dao.RdsUtils;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.dbutils.QueryRunner;
import org.apache.commons.dbutils.handlers.BeanListHandler;
import org.apache.commons.lang3.StringUtils;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.functions.ProcessFunction;
import org.apache.flink.util.Collector;
import org.apache.flink.util.OutputTag;
import redis.clients.jedis.Jedis;

import java.sql.Connection;
import java.text.MessageFormat;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.Objects;

/**
 * @Author: wengzx
 * @Date: 2025/4/24
 * @Desc:
 */
@Slf4j
public class ReportOnlineStatusHandler extends ProcessFunction<MessageContent, MessageContent> {

    public static final String T_IOTMO_DEVICE_INIT_SQL = "SELECT device_code as deviceCode, product_id as productId FROM {0} WHERE product_id IN (\"1401\", \"2334\", \"2289\", \"2337\", \"2338\", \"2629\", \"2628\") LIMIT ? OFFSET ?";
    public static final String T_IOTMO_PRODUCT_INIT_SQL = "SELECT product_id as productId, product_type as productType FROM {0} WHERE product_id IN (\"1401\", \"2334\", \"2289\", \"2337\", \"2338\", \"2629\", \"2628\") LIMIT ? OFFSET ?";

    private final OutputTag<EquipmentAlarmMonitorLogEnty> outTagStatus;
    private final RdsConfig rdsConfig;
    private final String tableName3;
    private final String tableName4;
    private String rds3SQL;
    private String rds4SQL;
    private RdsUtils rdsUtils;
    private final Map<String, Integer> iotpDevicecaches = new HashMap<>();
    private final Map<Integer, String> iotProductcaches = new HashMap<>();
    /**
     * 设备状态: 上线
     */
    public static final String ONLINE = "deviceOnline";

    /**
     * 设备状态: 下线
     */
    public static final String OFFLLINE = "deviceOffline";

    public ReportOnlineStatusHandler(OutputTag<EquipmentAlarmMonitorLogEnty> outTagStatus,RdsConfig rdsConfig,String rdsTableName3,String rdsTableName4) {
        this.outTagStatus = outTagStatus;
        this.rdsConfig = rdsConfig;
        this.tableName3 = rdsTableName3;
        this.tableName4 = rdsTableName4;

    }

    @Override
    public void open(Configuration parameters) throws Exception {
        rdsUtils = new RdsUtils(rdsConfig);
        rds3SQL = MessageFormat.format(T_IOTMO_DEVICE_INIT_SQL, tableName3);
        rds4SQL = MessageFormat.format(T_IOTMO_PRODUCT_INIT_SQL, tableName4);
        initDeviceCache();
        initProductCache();
    }

    @Override
    public void processElement(MessageContent mqttMsg, ProcessFunction<MessageContent, MessageContent>.Context ctx, Collector<MessageContent> out) throws Exception {
        JSONObject body = JSONObject.parseObject(mqttMsg.getData());
        if(body.isEmpty()){
            log.error("原始数据信息：{} 上报的数据为空," ,mqttMsg);
            return;
        }
        try {
            reportOnlineAlarm(body,ctx);
        } catch (Exception e) {
            log.error("原始数据信息：{}", mqttMsg);
            log.error(e.getMessage(), e);
        }
    }

    private void reportOnlineAlarm(JSONObject body, ProcessFunction<MessageContent, MessageContent>.Context ctx) {
        MsgKeyConstant msgKeyConstant = body.to(MsgKeyConstant.class);
        Map<String, Object> map = new HashMap<>();
        String notifyType = body.getString(msgKeyConstant.NOTIFY_TYPE);
        String deviceCode = body.getString(msgKeyConstant.DEVICE_IDENTIFIER);
        String eventTime = body.getString(msgKeyConstant.EVENT_TIME);
        String clientId = body.getString(msgKeyConstant.CLIENT_ID);
        long eventTimeLong = Long.parseLong(eventTime) * 1000;
        Integer projectId = iotpDevicecaches.get(deviceCode);
        String productType = iotProductcaches.get(projectId);
        if(projectId == null || StringUtils.isBlank(productType)) {
            return;
        }

        if (StrUtil.isEmpty(deviceCode)) {
            log.error("原始数据信息上报的deviceCode为空：{}",body);
            return;
        }
        if(!Objects.equals(notifyType,ONLINE) && !Objects.equals(notifyType,OFFLLINE)){
            log.error("原始数据信息上报的上下线值为空：{}",body);
            return;
        }
        //根据notifyType获取需要发送的指标
        String reportProxy = RuleReportProxyEnum.getReportProxy(notifyType);
        if (StrUtil.isBlank(reportProxy)) {
            log.info("告警数据为：【{}】的指标为空不予发送", msgKeyConstant);
            return;
        }
        EquipmentAlarmMonitorLogEnty equipmentAlarmMonitorLogEnty = new EquipmentAlarmMonitorLogEnty();
        //设置设备标识
        equipmentAlarmMonitorLogEnty.setDeviceCode(deviceCode);
        //设置指标
        equipmentAlarmMonitorLogEnty.setEventType(reportProxy);
        equipmentAlarmMonitorLogEnty.setProductId(projectId);
        equipmentAlarmMonitorLogEnty.setProductType(productType);
        String alarmCover;
        //获取到指标判断是恢复还是告警
        boolean recovery = RuleReportProxyEnum.isRecovery(notifyType);
        if (recovery) {
            //恢复自动处理
            alarmCover = EquipmentAlarmConstant.ALARM_COVER_AUTO;
            equipmentAlarmMonitorLogEnty.setPersonnelHandler("");
            equipmentAlarmMonitorLogEnty.setHandlingTime(DateUtil.date(eventTimeLong/1000));
        }else{
            //设置告警时间
            equipmentAlarmMonitorLogEnty.setAlarmTime(DateUtil.date(eventTimeLong/1000));
            //告警未处理
            alarmCover = EquipmentAlarmConstant.ALARM_COVER_UNPROCESSED;
            map.put("lastOfficeTime", eventTimeLong / 1000);
            map.put("offLineTime", 0);
            // 将 map 转换为 JSON 字符串
            String mapAsJson = JSON.toJSONString(map);
            equipmentAlarmMonitorLogEnty.setEventValue(mapAsJson);
        }
        equipmentAlarmMonitorLogEnty.setAlarmCover(alarmCover);
        equipmentAlarmMonitorLogEnty.setNetCode(EquipmentAlarmConstant.NET_CODE_INNER);
        equipmentAlarmMonitorLogEnty.setAlertLevel(EquipmentAlarmConstant.ALARM_LEVEL_CRITICAL);
        //消缺工具
        String blankingTool = "DEFECT_ELIMINATION";
        equipmentAlarmMonitorLogEnty.setBlankingTool(blankingTool);
        equipmentAlarmMonitorLogEnty.setMessageSuggestion(EquipmentProxyEnum.LAST_OFFLINE.getProxySuggest());
        equipmentAlarmMonitorLogEnty.setAlertDescription(StrUtil.format(EquipmentProxyEnum.LAST_OFFLINE.getProxyDesc(), StrUtil.SPACE));
        equipmentAlarmMonitorLogEnty.setAlertSource(EquipmentProxyEnum.LAST_OFFLINE.getProxyOrigin());
        equipmentAlarmMonitorLogEnty.setAlertDimensions("1");
        equipmentAlarmMonitorLogEnty.setNetCode(EquipmentAlarmConstant.NET_CODE_INNER);
        ctx.output(outTagStatus,equipmentAlarmMonitorLogEnty);
    }

    private void initDeviceCache() {
        try (Connection connection = rdsUtils.getConnection()) {
            // 创建 QueryRunner 对象
            QueryRunner queryRunner = new QueryRunner();
            // 设置分页参数
            int pageSize = 3000;
            int currentPage = 1;
            while (true) {
                log.info("加载{}缓存 {} {}", tableName3, currentPage, pageSize);
                int offset = (currentPage - 1) * pageSize;
                List<DeviceMoDO> query = queryRunner.query(connection, rds3SQL, new BeanListHandler<>(DeviceMoDO.class), pageSize, offset);
                if (query == null || query.isEmpty()) {
                    break;
                }
                for (DeviceMoDO deviceContainerDO : query) {
                    iotpDevicecaches.put(deviceContainerDO.getDeviceCode(), deviceContainerDO.getProductId());
                }
                currentPage++;
            }
            log.info("{} 缓存加载完成,大小为{}", tableName3,iotpDevicecaches.size());
        } catch (Exception e) {
            log.error(e.getMessage(), e);
        }
    }


    private void initProductCache() {
        try (Connection connection = rdsUtils.getConnection()) {
            // 创建 QueryRunner 对象
            QueryRunner queryRunner = new QueryRunner();
            // 设置分页参数
            int pageSize = 3000;
            int currentPage = 1;
            while (true) {
                log.info("加载{}缓存 {} {}", tableName4, currentPage, pageSize);
                int offset = (currentPage - 1) * pageSize;
                List<ProductDO> query = queryRunner.query(connection, rds4SQL, new BeanListHandler<>(ProductDO.class), pageSize, offset);
                if (query == null || query.isEmpty()) {
                    break;
                }
                for (ProductDO productDO : query) {
                    iotProductcaches.put(productDO.getProductId(), productDO.getProductType());
                }
                currentPage++;
            }
            log.info("{} 缓存加载完成,缓存大小为{}", tableName4,iotProductcaches.size());
        } catch (Exception e) {
            log.error(e.getMessage(), e);
        }
    }

    @Override
    public void close() throws Exception {
        if (rdsUtils != null) {
            rdsUtils.close();
        }
        super.close();
    }
}
